package com.kingsoft.dc.khaos.scheduler

import java.text.SimpleDateFormat
import java.util.concurrent.atomic.AtomicInteger

import com.kingsoft.dc.khaos.{DataSetChannel, KhaosConf, KhaosContext, SchemaChannel}
import com.kingsoft.dc.khaos.metadata.{Dependency, ModuleCommon, TaskInfo}
import com.kingsoft.dc.khaos.util.Logging

/**
  * Created by chengguangqing on 2019/6/1.
  */
class TaskSchedulerSparkImple(val kc:KhaosContext,
                                       val maxTaskFailures:Int
                             ) extends TaskScheduler with Logging{

  var _khaosScheduler: KhaosScheduler = null

  override def setKhaosScheduler(khaosScheduler: KhaosScheduler): Unit = {
    _khaosScheduler = khaosScheduler
  }

  val nextTaskId = new AtomicInteger(1)
  override def submitTasks(jobId: Int,
                           modules: List[ModuleCommon],
                           listener: ApplicationListener): Unit = {
    modules.foreach{ moduleCommon =>
      val taskId = nextTaskId.getAndIncrement();
      _khaosScheduler.pendingTasks += taskId
      val submissionTime = System.nanoTime()
      val taskInfo = TaskInfo(taskId,
        jobId,
        moduleCommon.id,
        moduleCommon.name,
        "SPARK",
        null,
        moduleCommon.dependence,
        Some(submissionTime))
      this._khaosScheduler.taskSubmitted(taskId, moduleCommon, taskInfo, listener)
    }
  }

  override def runTask(taskId: Int,
                       moduleCommon: ModuleCommon,
                       taskInfo: TaskInfo,
                       listener: ApplicationListener
                      ): Unit = {
    try{
      val sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
      log.info("-----------------------------------")
      log.info("START TIME:" + sdf.format(System.currentTimeMillis()))
      log.info("taskId is " + taskId)
      log.info("moduleId is " + moduleCommon.id)

      val moduleId = moduleCommon.id
      val clazz = moduleCommon.clazz

      //实例化组件
      val moduleInstance = Class.forName(clazz)
        .getDeclaredConstructor(classOf[String],classOf[KhaosContext])
        .newInstance(moduleId,kc)

      //添加依赖
      val addDependencies = moduleInstance.getClass()
        .getMethod("addDependencies",classOf[Dependency])
      moduleCommon.dataDependence.foreach{ dependency =>
        addDependencies.invoke(moduleInstance, dependency)
      }

      //添加出度
      val addTarget = moduleInstance.getClass()
        .getMethod("addTarget",classOf[Dependency])
      moduleCommon.degreeOutTarget.foreach{ dependency =>
        addTarget.invoke(moduleInstance, dependency)
      }

      //执行准备作
      val prepare = moduleInstance.getClass()
        .getMethod("prepare",classOf[KhaosConf],classOf[KhaosContext])
      prepare.invoke(moduleInstance,kc.conf,kc)

      //执行逻辑方法
      val execute = moduleInstance.getClass()
        .getMethod("execute",classOf[DataSetChannel])
      execute.invoke(moduleInstance,kc.dataSetChannel)

      //声明输出
      val declareOutputSchema = moduleInstance.getClass
        .getMethod("declareOutputSchema",classOf[SchemaChannel])
      declareOutputSchema.invoke(moduleInstance, kc.schemaChannel)

      //执行成功
      println("END TIME:" + sdf.format(System.currentTimeMillis()))
      val curTime = System.currentTimeMillis()
      taskInfo.completionTime = Some(curTime)

      this._khaosScheduler.taskResultCollector(taskInfo,listener)
    } catch {
      //21202F2938212B3E22272626252E434D
      case e:Exception =>{
        logError("Khaos task %d failed: %s, took %f s"
          .format(taskId, taskInfo.name, (System.nanoTime() - taskInfo.submissionTime.get) / 1e9),e)
        val callerStackTrace = Thread.currentThread().getStackTrace.tail
        e.setStackTrace(e.getStackTrace ++ callerStackTrace)
        this._khaosScheduler.taskFailed(taskInfo,Some(e),taskInfo.jobId,listener)
      }
    }
  }
}
